/**
 * 
 */
package io.gitee.falllee.dao.core.table;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.calcite.linq4j.Linq4j;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import io.gitee.falllee.dao.core.HeaderBuilder;
import io.gitee.falllee.dao.enums.SyncDatePrecisionEnum;
import io.gitee.falllee.dao.model.DaoConst;
import io.gitee.falllee.dao.model.DaoException;
import io.gitee.falllee.dao.persistence.DataMapper;
import io.gitee.falllee.dao.util.CommonUtils;
import io.gitee.falllee.dao.util.DaoUtils;
import io.gitee.falllee.dao.util.FilterPlus;

/**
 * @author batty
 * @date 2022-06-27
 *
 */
public class UpdateBuilder extends HeaderBuilder {
    
    private TableBuilder tableBuilder;
    
    private DataMapper dataMapper;
    
    private CacheBuilder cacheBuilder;
    
    private MessageBuilder messageBuilder;

    private String tableName;
    
    private String keyField;

    private Boolean ignoreNullData = true;
    
    private String updateFields;
    
    private Integer updateBatchMaxSize = 500;
    
    /**
     * 是否同步表（同步表的固定字段取实体传递的值）
     */
    private boolean syncTable;

    /**
     * 同步时间字段（如果syncTable=true，则可以指定此字段为同步时间字段，取数据库当前时间）
     */
    private String syncDateField;
    
    /**
     * 同步时间精度
     */
    private SyncDatePrecisionEnum syncDatePrecision;
    
    UpdateBuilder(TableBuilder parent) {
        super(parent.getDaoConfig());
        this.tableBuilder = parent;
        this.dataMapper = parent.getDataMapper();
        this.cacheBuilder = parent.cacheBuilder();
        this.messageBuilder = parent.messageBuilder();
        this.tableName = parent.getTableName();
        this.keyField = parent.getKeyField();
        this.syncTable = parent.syncTable();
        this.syncDateField = parent.syncDateField();
        this.syncDatePrecision = parent.syncDatePrecision();
    }
    
    public UpdateBuilder header(String key, String value) {
        if (headerParamsMap == null) {
            headerParamsMap = new HashMap<>(16);
        }
        headerParamsMap.put(key, value);
        return this;
    }
    
    public UpdateFilterBuilder filter(FilterPlus filter) {
        return new UpdateFilterBuilder(this, daoConfig, filter);
    }
    
    public UpdateBuilder allowUpdateNullData() {
        this.ignoreNullData = false;
        return this;
    }
    
    public UpdateBuilder updateFields(String updateFields) {
        this.updateFields = updateFields;
        return this;
    }
    
    public UpdateBuilder updateBatchMaxSize(Integer updateBatchMaxSize) {
        this.updateBatchMaxSize = updateBatchMaxSize;
        return this;
    }
    
    public Boolean update(Map<String, Object> t) {
        if (t == null) {
            return false;
        }
        Map<String, Object> map = new HashMap<>(16);
        map.put(DaoConst.DB_TYPE, daoConfig.getDbType());
        Object keyValue = setUpdateFieldAndValue(t, map);
        map.put(DaoConst.TABLE_NAME, tableName);
        if (!syncTable) {
            setEnableFlag(map, false);
            setUpdateInfo(map);
        } else {
            setSyncDate(map, syncDateField, syncDatePrecision.getPrecision());
        }
        Integer res = dataMapper.update(map);
        if (res > 0) {
            // 如果启用缓存，删除缓存数据
            cacheBuilder.clearCache(keyValue);
            // 发送Kafka消息
            messageBuilder.sendMessage(keyValue, DaoConst.UPDATE);
        }
        return res >= 0;
    }

    public Boolean updateBatch(List<Map<String, Object>> list) {
        List<Map<String, Object>> tempList = Linq4j.asEnumerable(list).where(p -> !ObjectUtils.isEmpty(p)).distinct()
                .toList();
        if (CollectionUtils.isEmpty(tempList)) {
            return false;
        }
        Map<String, Object> genMap = new HashMap<>(16);
        genMap.put(DaoConst.DB_TYPE, daoConfig.getDbType());
        genMap.put(DaoConst.TABLE_NAME, tableName);
        genMap.put("id", keyField);
        if (!syncTable) {
            setEnableFlag(genMap, false);
            setUpdateInfo(genMap);
        } else {
            setSyncDate(genMap, syncDateField, syncDatePrecision.getPrecision());
        }
        int total = tempList.size();
        int batchSize = updateBatchMaxSize;
        // 执行次数
        int num = total / batchSize;
        // 最后一次剩余记录数
        int left = total % batchSize;
        if (left > 0) {
            num += 1;
        }
        String[] columnNames = getUpdateColumns(tempList.get(0));
        List<Object> keyValueList = new ArrayList<>();
        Integer res = 0;
        for (int i = 0; i < num; i++) {
            List<Map<String, Object>> subList = CommonUtils.getSubList(tempList, batchSize, i);
            // 主键值的集合
            res += updateBatchSubList(genMap, columnNames, keyValueList, subList);
        }
        if (res > 0) {
            // 清除实体缓存
            cacheBuilder.clearCache(keyValueList);
            // 发送Kafka消息
            messageBuilder.sendMessage(keyValueList, DaoConst.UPDATE_LIST);
        }
        return res >= 0;
    }

    private String[] getUpdateColumns(Map<String, Object> t) {
        // 业务自定义固定字段
        List<String> fixFieldList = listFixField();
        List<String> configFields = getComponentConfigFields(syncTable, syncDateField);
        List<String> columnList = new ArrayList<>();
        for (String columName : t.keySet()) {
            // 固定字段在sql中处理
            if (configFields.contains(columName) || fixFieldList.contains(columName.toLowerCase())) {
                continue;
            }
            columnList.add(columName);
        }
        return columnList.toArray(new String[] {});
    }

    private Object setUpdateFieldAndValue(Map<String, Object> t, Map<String, Object> map) {
        Object keyValue = t.get(keyField);
        if (ObjectUtils.isEmpty(keyValue)) {
            throw new DaoException("keyword value must be not null");
        }
        String fieldValue = "";
        // 主键作为更新条件
        String filter = getParamExpression(keyField);
        StringBuilder sbFieldValue = new StringBuilder();
        // 业务自定义固定字段
        List<String> fixFieldList = listFixField();
        String[] updateFieldArr = new String[0];
        if (StringUtils.hasLength(updateFields)) {
            updateFieldArr = updateFields.toLowerCase().split(DaoConst.COMMA);
        }
        List<String> updateFieldList = Arrays.asList(updateFieldArr);
        Iterator<Entry<String, Object>> iterator = t.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, Object> entry = iterator.next();
            String columnName = entry.getKey();
            Object objVal = entry.getValue();
            String colQuotes = getQuoteField(columnName);
            // 如果指定了字段，无论是否空值均按指定字段更新
            // 业务自定义字段不更新
            boolean notUpdate = !ObjectUtils.isEmpty(updateFieldArr)
                    && !updateFieldList.contains(columnName.toLowerCase())
                    || ObjectUtils.isEmpty(updateFieldArr) && ignoreNullData && ObjectUtils.isEmpty(objVal)
                    || fixFieldList.contains(columnName.toLowerCase()) || sbFieldValue.toString().contains(colQuotes);
            if (notUpdate) {
                continue;
            }
            // 组装set语句，非主键字段
            if (!keyField.equals(columnName)) {
                sbFieldValue.append(getParamExpression(columnName));
                sbFieldValue.append(DaoConst.COMMA);
            }
            // 赋值
            map.put(columnName, objVal);
        }
        if (sbFieldValue.length() > 0) {
            fieldValue = sbFieldValue.substring(0, sbFieldValue.length() - 1);
        } else {
            throw new DaoException(DaoConst.FIELD_CAN_NOT_BE_EMPTY);
        }
        map.put("fieldValue", fieldValue);
        map.put(DaoConst.FILTER, filter);
        map.computeIfAbsent(keyField, p-> keyValue);
        return keyValue;
    }

    private Integer updateBatchSubList(Map<String, Object> genMap, String[] columnNames, List<Object> keyValueList,
            List<Map<String, Object>> subList) {
        Integer res;
        List<Object> subKeyValueList = new ArrayList<>();
        for (Map<String, Object> t : subList) {
            Object keyValue = t.get(keyField);
            subKeyValueList.add(keyValue);
        }
        keyValueList.addAll(subKeyValueList);
        Map<String, List<Map<Object, Object>>> columnCaseMap = new HashMap<>(16);
        for (String columnName : columnNames) {
            List<Map<Object, Object>> caseList = getCaseValue(subList, columnName);
            if (!CollectionUtils.isEmpty(caseList)) {
                columnCaseMap.put(getQuoteField(columnName), caseList);
            }
        }
        res = dataMapper.updateBatch(genMap, subKeyValueList, columnCaseMap);
        return res;
    }

    private List<Map<Object, Object>> getCaseValue(List<Map<String, Object>> subList, String columnName) {
        List<Map<Object, Object>> caseList = new ArrayList<>();
        for (Map<String, Object> t : subList) {
            Object keyValue = t.get(keyField);
            Object objVal = t.get(columnName);
            // 主键必须非空；如果指定了更新字段，无论空值均需要更新
            boolean notUpdate = ObjectUtils.isEmpty(keyValue)
                    || ObjectUtils.isEmpty(updateFields) && ObjectUtils.isEmpty(objVal) && ignoreNullData;
            if (notUpdate) {
                continue;
            }
            Map<Object, Object> caseValues = new HashMap<>(16);
            caseValues.put(keyValue, objVal);
            caseList.add(caseValues);
        }
        return caseList;
    }

    /**
     * 根据Map的键值对更新（更新字段空值推荐此方法）
     * 
     * @param headerParamsMap 请求头列表
     * @param fieldValueMap   更新字段的键值对（根据条件更新，无需传主键的键值对）
     * @param filter          条件
     * @param clazz           实体类型
     * @return 是否成功
     */
    Boolean update(Map<String, Object> fieldValueMap, FilterPlus filterPlus) {
        if (ObjectUtils.isEmpty(fieldValueMap)) {
            return false;
        }
        String paramFilter = filterPlus.getParamFilter();
        if (!StringUtils.hasLength(paramFilter)) {
            throw new DaoException("filter must not be empty");
        }
        Map<String, Object> map = filterPlus.getParamMap();
        map.put(DaoConst.DB_TYPE, daoConfig.getDbType());
        map.put(DaoConst.TABLE_NAME, tableName);
        map.put(DaoConst.FILTER, paramFilter);
        setUpdateKeyValueMap(fieldValueMap, keyField, map);
        setFixInfo(map);
        // 先查待更新的数量，分页进行更新
        long count = dataMapper.selectCount(map);
        long num = count / DaoConst.SELECT_MAX_SIZE;
        long left = count % DaoConst.SELECT_MAX_SIZE;
        num = left > 0 ? num + 1 : num;
        long rows = DaoConst.SELECT_MAX_SIZE;
        Integer res = 0;
        for (long i = 0; i < num; i++) {
            long end = i == num - 1 ? count : (i + 1) * rows;
            map.put(DaoConst.FILTER, paramFilter);
            List<Object> keyList = tableBuilder.getKeyList(map, keyField, i * rows, rows, end);
            String filter = getKeyListFilter(keyList, keyField, map);
            map.put(DaoConst.FILTER, filter);
            res += dataMapper.update(map);
            if (res > 0) {
                // 如果启用缓存，删除缓存数据
                cacheBuilder.clearCache(keyList);
                // 发送Kafka消息
                messageBuilder.sendMessage(keyList, DaoConst.UPDATE_LIST);
            }
        }
        return res >= 0;
    }

    private void setUpdateKeyValueMap(Map<String, Object> fieldValueMap, String keyField, Map<String, Object> map) {
        StringBuilder sbFieldValue = new StringBuilder();
        Iterator<Entry<String, Object>> it = fieldValueMap.entrySet().iterator();
        while (it.hasNext()) {
            Entry<String, Object> entry = it.next();
            String columnName = entry.getKey();
            // 主键禁止更新
            if (columnName.equalsIgnoreCase(keyField)) {
                continue;
            }
            // 组装set语句，非主键字段，如果map包含字段可能是filter已经存在，需要修改参数字段
            String paramName = columnName;
            if (map.containsKey(columnName)) {
                paramName = getParamName();
            }
            sbFieldValue.append(getParamExpression(columnName, paramName));
            sbFieldValue.append(DaoConst.COMMA);
            // 参数赋值
            map.put(paramName, entry.getValue());
        }
        String fieldValue = "";
        if (sbFieldValue.length() > 0) {
            fieldValue = sbFieldValue.substring(0, sbFieldValue.length() - 1);
        } else {
            throw new DaoException(DaoConst.FIELD_CAN_NOT_BE_EMPTY);
        }
        map.put("fieldValue", fieldValue);
    }

    private void setFixInfo(Map<String, Object> map) {
        if (!syncTable) {
            // 设置业务固定字段（从请求头获取）
            Map<String, Object> fixFieldValue = getFixedFieldValue();
            DaoUtils.setFixFieldValue(fixFieldValue, map);
            setUpdateInfo(map);
        } else {
            setSyncDate(map, syncDateField, syncDatePrecision.getPrecision());
        }
    }
}
